package com.zx.mes.rocketmq.hyl.quickstart;

import com.zx.mes.rocketmq.hyl.constant.Const;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

/**
 * 选择某个队列进行消费
 *
 * @author 华云龙
 * @date 2018-11-30
 */
public class SelectorQueuePorducer {


    private static final int INT = 5;
    private static final String TEST_QUICK_TOPIC = "test_quick_topic";

    public static void send() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {

        String producerGroup = "test_quick_producer_name";
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(Const.NAMESRV_ADDR_SINGLE);

        producer.start();

        for (int i = 0; i <INT ; i++) {
            Message message = new Message(TEST_QUICK_TOPIC, "selector tag", "selector key:" + i, ("选择队列进行消费:" + i).getBytes());
            /// public SendResult send(Message msg, MessageQueueSelector selector, Object arg)
            /// arg 为MessageQueueSelector接口中重写函数中参数arg
            SendResult result = producer.send(message, (mqs, msg, arg) -> mqs.get((Integer) arg), 2);
            System.out.println(result);
        }

        producer.shutdown();
    }
}
